//
// Created By: Xiaowen.Zhu 2021/06/18
//
#include "cybertron/core/FileDownloader.hpp"
#include <sys/stat.h>
#define CURL_STATICLIB
#include <curl/multi.h>
#include <miniz.h>
#include <unordered_map>
#include <unordered_set>
#include <map>
#include <iostream>
#include <mutex>
#include <queue>
#include <thread>

namespace FDTM_IMPL {

#ifdef CYBERTRON_WIN
#define FILE_SEP '\\'
#else
#define FILE_SEP '/'
#endif

    class Util {
    public:
        static std::string get_filename_from_url(char const *url) {
            char const *u = url;

            /* Remove "http(s)://" */
            u = strstr(u, "://");
            if (u) {
                u += strlen("://");
            }

            u = strrchr(u, '/');

            /* Remove last '/' */
            u++;

            return std::string(u);
        }

        static std::string fix_dir(const char *dir) {
            if (dir[strlen(dir) - 1] != FILE_SEP) {
                return std::string(dir) + FILE_SEP;
            } else {
                return std::string(dir);
            }
        }

        static FILE *open_file(const char *path, const char *mode, bool makedir = false) {
            if (makedir) {
                mk_dir(path);
            }
#ifdef CYBERTRON_WIN
            WCHAR w_path[MAX_PATH];
            WCHAR w_mode[16];
            MultiByteToWideChar(CP_UTF8, 0, path, int(strlen(path) + 1), w_path, MAX_PATH);
            MultiByteToWideChar(CP_UTF8, 0, mode, int(strlen(mode) + 1), w_mode, 16);
            return _wfopen(w_path, w_mode);
#else
            return fopen(path, mode);
#endif
        }

        static int remove_file(const char *path) {
#ifdef CYBERTRON_WIN
            WCHAR name[MAX_PATH];
            MultiByteToWideChar(CP_UTF8, 0, path, int(strlen(path) + 1), name, MAX_PATH);
            return _wremove(name);
#else
            return remove(path);
#endif
        }

        static int rename_file(const char *src_path, const char *dst_path, bool makedir) {
            if (makedir) {
                mk_dir(dst_path);
            }

#ifdef CYBERTRON_WIN
            WCHAR old_name[MAX_PATH];
            WCHAR new_name[MAX_PATH];
            MultiByteToWideChar(CP_UTF8, 0, src_path, int(strlen(src_path) + 1), old_name, MAX_PATH);
            MultiByteToWideChar(CP_UTF8, 0, dst_path, int(strlen(dst_path) + 1), new_name, MAX_PATH);
            return _wrename(old_name, new_name);
#else
            return rename(src_path, dst_path);
#endif
        }

        static int mk_dir(const char *path) {
            std::string full_path = std::string(path);
            std::vector<std::string> v;

            std::string prefix = "";
            if (path != NULL && path[0] == '/') //for mac and linux system
                prefix = "/";

            while (1) {
                size_t found = full_path.find_first_of("/\\");
                if (found == -1)
                    break;
                else {
                    if (found > 0) {
                        //printf("%s\n", full_path.substr(0, found).c_str());
                        v.push_back(prefix + full_path.substr(0, found));
                        prefix = "";
                    }
                    full_path = full_path.substr(found + 1);
                }
            }

            std::string dir;
            for (size_t i = 0; i < v.size(); i++) {
                dir += v.at(i);
#ifdef CYBERTRON_WIN
                WCHAR name[MAX_PATH];
                MultiByteToWideChar(CP_UTF8, 0, dir.c_str(), int(dir.length() + 1), name, MAX_PATH);
                struct _stat64 s;
                if (_wstat64(name, &s) != 0) {
                    if (errno == ENOENT)
                        _wmkdir(name);
                    else return -1;
                }
#else
                struct stat s;
                if (stat(dir.c_str(), &s) != 0)
                {
                    if (errno == ENOENT)
                        mkdir(dir.c_str(), 0755);
                    else return -1;
                }
#endif
                dir += "/";
            }

            return 0;
        }

        static int unzip(std::string zip_file_path, std::string dst_dir) {
            mz_zip_archive zip_archive;
            memset(&zip_archive, 0, sizeof(zip_archive));

            auto status = mz_zip_reader_init_file(&zip_archive, zip_file_path.c_str(), 0);
            if (!status) return -1;
            int fileCount = (int) mz_zip_reader_get_num_files(&zip_archive);
            if (fileCount == 0) {
                mz_zip_reader_end(&zip_archive);
                return -1;
            }
            mz_zip_archive_file_stat file_stat;
            if (!mz_zip_reader_file_stat(&zip_archive, 0, &file_stat)) {
                mz_zip_reader_end(&zip_archive);
                return -1;
            }

            for (int i = 0; i < fileCount; i++) {
                if (!mz_zip_reader_file_stat(&zip_archive, i, &file_stat)) continue;

                std::string dst_file_path = dst_dir + std::string(file_stat.m_filename);

                if (mz_zip_reader_is_file_a_directory(&zip_archive, i)) {
                    mk_dir(dst_file_path.c_str());
                    continue;
                }

                FILE *file_handle = open_file(dst_file_path.c_str(), "wb", true);
                if (file_handle == nullptr) {
                    mz_zip_reader_end(&zip_archive);
                    return -1;
                }

                auto rc = mz_zip_reader_extract_to_callback(&zip_archive, i, &unzip_write_func, file_handle, 0);
                fclose(file_handle);
                if (!rc) {
                    mz_zip_reader_end(&zip_archive);
                    return -1;
                }
            }
            mz_zip_reader_end(&zip_archive);
            return 0;
        }

    private:
        static size_t unzip_write_func(void *pOpaque, mz_uint64 file_ofs, const void *pBuf, size_t n) {
            FILE *file_handle = (FILE *) pOpaque;
            return fwrite(pBuf, 1, n, file_handle);
        }
    };

    class Task {
    public:
        Task(std::string url)
                : _url(url),
                  _file_handle(nullptr) {

        }

        Task(std::string url, std::string md5sum, std::string dst_dir, bool decompress,
             cybertron::FDT_STATUS_CB user_cb, cybertron::FDT_STATUS_CB manager_cb)
                : _url(url),
                  _md5sum(md5sum),
                  _decompress(decompress),
                  _head_done(false),
                  _to_delete(false),
                  _stage(cybertron::FDT_STAGE_QUEUING),
                  _retry_count(0),
                  _user_cb(user_cb),
                  _manager_cb(manager_cb),
                  _file_handle(nullptr),
                  _last_error(cybertron::FDT_ERROR_OK),
                  _total_size(0),
                  _done_size(0),
                  _time_last_update_status(0),
                  _curl(nullptr) {
            _dst_dir = Util::fix_dir(dst_dir.c_str());
            _dst_path = _dst_dir + Util::get_filename_from_url(url.c_str());
        }

        ~Task() {
            close_file();
        }

        int open_file() {
            _file_handle = Util::open_file(_dst_path.c_str(), "ab", true);
            if (_file_handle != nullptr) {
#ifdef WIN32
                _done_size = _ftelli64(_file_handle);
#else
                _done_size = ftello64(_file_handle);
#endif
                if (_done_size < 0) {
                    _done_size = 0;
                }
            }
            return _file_handle == nullptr ? -1 : 0;
        }

        void close_file() {
            if (_file_handle) {
                fclose(_file_handle);
                _file_handle = nullptr;
            }
        }

        void callback_status(bool force = false) {
            time_t now = time(NULL);
            if (!force && now - _time_last_update_status < 1) {
                return;
            }

            _time_last_update_status = now;

            cybertron::FDT_STATUS status;
            status._stage = _stage;
            status._last_error = _last_error;
            status._total_size = _total_size;
            status._done_size = _done_size;
            status._retry_count = _retry_count;

            if (_user_cb) {
                _user_cb(_url, status);
            }

            if (_manager_cb) {
                _manager_cb(_url, status);
            }
        }

        std::string _url;
        std::string _md5sum;
        std::string _dst_path;
        std::string _dst_dir;
        bool _decompress;
        bool _head_done;
        bool _to_delete;
        cybertron::FDT_STAGE _stage;
        cybertron::FDT_ERROR _last_error;
        int _retry_count;
        size_t _total_size;
        size_t _done_size;
        FILE *_file_handle;
        cybertron::FDT_STATUS_CB _user_cb;
        cybertron::FDT_STATUS_CB _manager_cb;
        time_t _time_last_update_status;
        CURL *_curl;
    };

    enum FDT_CMD {
        FDT_CMD_ADD,
        FDT_CMD_DELETE,
        FDT_CMD_UNZIP,
        FDT_CMD_UNZIP_RESULT,
    };

    struct Cmd {
        FDT_CMD _cmd;
        Task *_task;
    };

    class Manager {
    public:
        static Manager *instance() {
            return _instance;
        }

        int start(int max_concurrent_task_num, cybertron::FDT_STATUS_CB callback) {
//#ifdef CYBERTRON_WIN
//            WSADATA wsaData;
//            WORD wVersionRequested = MAKEWORD(2, 2);
//            WSAStartup(wVersionRequested, &wsaData);
//#endif
            _max_concurrent_task_num = max_concurrent_task_num;
            _cb = callback;
            _quit = false;

            _download_thread = std::thread(&Manager::download_thread_func, this);
            _unzip_thread = std::thread(&Manager::unzip_thread_func, this);
            return 0;
        }

        int stop() {
            _quit = true;
            _download_thread.join();
            _unzip_thread.join();
            return 0;
        }

        int add_task(std::string url, std::string md5sum, std::string dst_path, bool decompress,
                     cybertron::FDT_STATUS_CB callback) {
            Cmd cmd;
            cmd._cmd = FDT_CMD_ADD;
            cmd._task = new Task(url, md5sum, dst_path, decompress, callback == nullptr ? _cb : callback,
                                 &Manager::on_status_change);
            _cmd_queue_mutex.lock();
            _cmd_queue.push(cmd);
            _cmd_queue_mutex.unlock();
            return 0;
        }

        int delete_task(std::string url) {
            Cmd cmd;
            cmd._cmd = FDT_CMD_DELETE;
            cmd._task = new Task(url);
            _cmd_queue_mutex.lock();
            _cmd_queue.push(cmd);
            _cmd_queue_mutex.unlock();
            return 0;
        }

        int task_status(std::string url, cybertron::FDT_STATUS &status) {
            std::lock_guard<std::mutex> lock(_task_status_mutex);
            auto iter = _task_status.find(url);
            if (iter == _task_status.end()) {
                return -1;
            } else {
                status = iter->second;
                return 0;
            }
        }

    private:
        static int on_status_change(std::string url, cybertron::FDT_STATUS status) {
            std::lock_guard<std::mutex> lock(instance()->_task_status_mutex);
            instance()->_task_status[url] = status;
            return 0;
        }


        static size_t curl_on_recv_cb(char *ptr, size_t size, size_t nmemb, void *userdata) {
            Task *task = (Task *) userdata;
            size_t n = fwrite(ptr, size, nmemb, task->_file_handle);

            if (n > 0) {
                task->_done_size += n;
            }
            task->callback_status(false);
            return n;
        }


        void run_task(Task *task, CURLM *curl_multi) {
            task->_stage = cybertron::FDT_STAGE_DONWLOADING;
            if (task->_last_error != cybertron::FDT_ERROR_OK) {
                task->_retry_count++;
                task->_last_error = cybertron::FDT_ERROR_OK;
                task->callback_status(true);
            }

            CURL *curl = curl_easy_init();
            curl_multi_add_handle(curl_multi, curl);
            task->_curl = curl;

            curl_easy_setopt(curl, CURLOPT_CONNECTTIMEOUT, 30L);
            curl_easy_setopt(curl, CURLOPT_SSL_VERIFYPEER, 0L);
            curl_easy_setopt(curl, CURLOPT_SSL_VERIFYHOST, 0L);
            curl_easy_setopt(curl, CURLOPT_LOW_SPEED_LIMIT, 1024L);
            curl_easy_setopt(curl, CURLOPT_LOW_SPEED_TIME, 5L);
            curl_easy_setopt(curl, CURLOPT_PRIVATE, task);
            curl_easy_setopt(curl, CURLOPT_URL, task->_url.c_str());

            if (task->_head_done) {
                curl_easy_setopt(curl, CURLOPT_WRITEDATA, task);
                curl_easy_setopt(curl, CURLOPT_BUFFERSIZE, CURL_MAX_READ_SIZE);
                //curl_easy_setopt(curl, CURLOPT_TIMEOUT, 3600);
                curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, Manager::curl_on_recv_cb);

                if (task->_done_size > 0) {
                    char buf[64];
                    //sprintf(buf, "%zu-%zu", task->_done_size, task->_total_size - 1);
                    sprintf(buf, "%zu-", task->_done_size);
                    curl_easy_setopt(curl, CURLOPT_RANGE, buf);
                }
            } else {
                curl_easy_setopt(curl, CURLOPT_TIMEOUT, 30L);
                curl_easy_setopt(curl, CURLOPT_NOBODY, 1L);
            }
        }

        void delete_task(Task *task, CURLM *curl_multi) {
            task->_stage = cybertron::FDT_STAGE_DELETED;
            task->callback_status(true);

            if (task->_curl) {
                curl_multi_remove_handle(curl_multi, task->_curl);
                curl_easy_cleanup(task->_curl);
            }
            delete task;
        }

        void unzip_thread_func() {
            while (!_quit) {
                Task *task = nullptr;
                _wait_for_unzip_tasks_mutex.lock();
                if (!_wait_for_unzip_tasks.empty()) {
                    auto iter = _wait_for_unzip_tasks.begin();
                    task = iter->second;
                    _wait_for_unzip_tasks.erase(iter);
                }
                _wait_for_unzip_tasks_mutex.unlock();

                if (!task) {
                    std::this_thread::sleep_for(std::chrono::milliseconds(100));
                    continue;
                }

                std::string url = task->_url;
                std::string dst_path = task->_dst_path;
                std::string dst_dir = task->_dst_dir;
                delete task;

                Cmd cmd;
                cmd._cmd = FDT_CMD_UNZIP_RESULT;
                cmd._task = new Task(url);
                _cmd_queue_mutex.lock();
                _cmd_queue.push(cmd);
                _cmd_queue_mutex.unlock();

                int rc = Util::unzip(dst_path, dst_dir);

                cmd._task = new Task(url);
                cmd._task->_last_error = rc == 0 ? cybertron::FDT_ERROR_OK : cybertron::FDT_ERROR_UNZIP;
                _cmd_queue_mutex.lock();
                _cmd_queue.push(cmd);
                _cmd_queue_mutex.unlock();
            }
        }

        void download_thread_func() {
            curl_global_init(CURL_GLOBAL_ALL);
            CURLM *curl_multi = curl_multi_init();
            int still_running = 0;
            while (!_quit) {
                //handle command
                _cmd_queue_mutex.lock();
                while (!_cmd_queue.empty()) {
                    Cmd cmd = _cmd_queue.front();
                    _cmd_queue.pop();
                    switch (cmd._cmd) {
                        case FDT_CMD_ADD: {
                            if (_tasks.count(cmd._task->_url) == 0) {
                                if (cmd._task->open_file() == 0) {
                                    _tasks[cmd._task->_url] = cmd._task;
                                    _queuing_tasks.insert(std::make_pair(time(NULL), cmd._task));
                                    cmd._task->_last_error = cybertron::FDT_ERROR_OK;
                                    cmd._task->_stage = cybertron::FDT_STAGE_QUEUING;
                                    cmd._task->callback_status(true);
                                } else {
                                    cmd._task->_last_error = cybertron::FDT_ERROR_FILE;
                                    cmd._task->_stage = cybertron::FDT_STAGE_FAILED;
                                    cmd._task->callback_status(true);
                                    std::cout << "FDT: can't open destination file\n";
                                    delete cmd._task;
                                }
                            } else {
                                cmd._task->_last_error = cybertron::FDT_ERROR_DUPLICATE;
                                cmd._task->_stage = cybertron::FDT_STAGE_FAILED;
                                cmd._task->callback_status(true);
                                delete cmd._task;
                            }
                            break;
                        }
                        case FDT_CMD_DELETE: {
                            auto iter = _tasks.find(cmd._task->_url);
                            if (iter != _tasks.end()) {
                                iter->second->_to_delete = true;
                            }
                            delete cmd._task;
                            break;
                        }
                        case FDT_CMD_UNZIP_RESULT: {
                            auto iter = _tasks.find(cmd._task->_url);
                            if (iter != _tasks.end()) {
                                Task *task = iter->second;
                                if (task->_stage == cybertron::FDT_STAGE_WAITING_FOR_UNZIP) {
                                    task->_stage = cybertron::FDT_STAGE_UNZIPING;
                                } else {
                                    if (cmd._task->_last_error == cybertron::FDT_ERROR_OK) {
                                        task->_last_error = cybertron::FDT_ERROR_OK;
                                        task->_stage = cybertron::FDT_STAGE_COMPLETED;
                                    } else {
                                        task->_last_error = cybertron::FDT_ERROR_UNZIP;
                                        task->_stage = cybertron::FDT_STAGE_FAILED;
                                    }
                                }
                                task->callback_status(true);
                            }
                            delete cmd._task;
                            break;
                        }
                        default:
                            delete cmd._task;
                            break;
                    }
                }
                _cmd_queue_mutex.unlock();

                //delete tasks
                for (auto iter = _tasks.begin(); iter != _tasks.end();) {
                    Task *task = iter->second;
                    if (task->_to_delete) {
                        _tasks.erase(iter++);
                        switch (task->_stage) {
                            case cybertron::FDT_STAGE_QUEUING:
                                for (auto it = _queuing_tasks.begin(); it != _queuing_tasks.end(); it++) {
                                    if (it->second == task) {
                                        _queuing_tasks.erase(it);
                                        break;
                                    }
                                }
                                break;
                            case cybertron::FDT_STAGE_DONWLOADING:
                                _downloading_tasks.erase(task);
                                break;
                            case cybertron::FDT_STAGE_WAITING_FOR_RETRY:
                                for (auto it = _wait_for_retry_tasks.begin(); it != _wait_for_retry_tasks.end(); it++) {
                                    if (it->second == task) {
                                        _wait_for_retry_tasks.erase(it);
                                        break;
                                    }
                                }
                                break;
                            case cybertron::FDT_STAGE_WAITING_FOR_UNZIP:
                                _wait_for_unzip_tasks_mutex.lock();
                                for (auto it = _wait_for_unzip_tasks.begin(); it != _wait_for_unzip_tasks.end(); it++) {
                                    if (it->second->_url == task->_url) {
                                        delete it->second;
                                        _wait_for_unzip_tasks.erase(it);
                                        break;
                                    }
                                }
                                _wait_for_unzip_tasks_mutex.unlock();
                                break;
                            case cybertron::FDT_STAGE_UNZIPING:
                            case cybertron::FDT_STAGE_FAILED:
                            case cybertron::FDT_STAGE_DELETED:
                            case cybertron::FDT_STAGE_COMPLETED:
                            default:
                                break;
                        }
                        _task_status_mutex.lock();
                        _task_status.erase(task->_url);
                        _task_status_mutex.unlock();
                        delete_task(task, curl_multi);
                    } else {
                        iter++;
                    }
                }

                //select task to run
                while (_downloading_tasks.size() < _max_concurrent_task_num) {
                    if (!_wait_for_retry_tasks.empty()) {
                        auto iter = _wait_for_retry_tasks.begin();
                        Task *task = iter->second;
                        time_t time_to_retry = iter->first;
                        if (time_to_retry < ::time(NULL)) {
                            _wait_for_retry_tasks.erase(iter);
                            _downloading_tasks.insert(task);
                            run_task(task, curl_multi);
                            continue;
                        }
                    }

                    if (!_queuing_tasks.empty()) {
                        auto iter = _queuing_tasks.begin();
                        Task *task = iter->second;
                        _queuing_tasks.erase(iter);
                        _downloading_tasks.insert(task);
                        run_task(task, curl_multi);
                        continue;
                    }
                    break;
                }

                //curl multi
                struct timeval timeout;
                timeout.tv_sec = 0;
                timeout.tv_usec = 100 * 1000;

                fd_set fdread;
                fd_set fdwrite;
                fd_set fdexcep;
                int maxfd = -1;

                long curl_timeo = -1;

                curl_multi_timeout(curl_multi, &curl_timeo);
                if (curl_timeo >= 0) {
                    timeout.tv_sec = curl_timeo / 1000;
                    if (timeout.tv_sec > 1)
                        timeout.tv_sec = 1;
                    else
                        timeout.tv_usec = (curl_timeo % 1000) * 1000;
                }

                FD_ZERO(&fdread);
                FD_ZERO(&fdwrite);
                FD_ZERO(&fdexcep);

                curl_multi_fdset(curl_multi, &fdread, &fdwrite, &fdexcep, &maxfd);

                int rc;
                if (maxfd == -1) {
                    rc = 0;
                    std::this_thread::sleep_for(std::chrono::milliseconds(100));
                } else {
                    /* Note that on some platforms 'timeout' may be modified by select().
                       If you need access to the original value save a copy beforehand. */
                    rc = select(maxfd + 1, &fdread, &fdwrite, &fdexcep, &timeout);
                }

                switch (rc) {
                    case -1:
                        //max_select = 0;
                        /* select error */
                        break;
                    case 0: /* timeout */
                    default: /* action */
                        curl_multi_perform(curl_multi, &still_running);
                        break;
                }

                CURLMsg *msg;
                int msgs_left;
                do {
                    msg = curl_multi_info_read(curl_multi, &msgs_left);
                    if (msg) {
                        Task *task;
                        curl_easy_getinfo(msg->easy_handle, CURLINFO_PRIVATE, &task);
                        switch (msg->data.result) {
                            case CURLE_OK:
                                long response_code;
                                curl_easy_getinfo(msg->easy_handle, CURLINFO_RESPONSE_CODE, &response_code);
                                if (response_code == 200 || response_code == 206) {
                                    task->_last_error = cybertron::FDT_ERROR_OK;
                                    if (!task->_head_done) {
                                        curl_off_t content_length;
                                        curl_easy_getinfo(msg->easy_handle, CURLINFO_CONTENT_LENGTH_DOWNLOAD_T,
                                                          &content_length);
                                        task->_total_size = content_length;
                                        task->_head_done = true;
                                    } else { //download done
                                        task->close_file();
                                        if (task->_decompress) {
                                            task->_stage = cybertron::FDT_STAGE_WAITING_FOR_UNZIP;
                                        } else {
                                            task->_stage = cybertron::FDT_STAGE_COMPLETED;
                                        }
                                    }
                                } else {
                                    task->_last_error = cybertron::FDT_ERROR_SERVER;
                                    task->_stage = cybertron::FDT_STAGE_WAITING_FOR_RETRY;
                                }
                                break;
                            case CURLE_ABORTED_BY_CALLBACK:
                                task->_last_error = cybertron::FDT_ERROR_FILE;
                                task->_stage = cybertron::FDT_STAGE_WAITING_FOR_RETRY;
                                break;
                            default:
                                task->_last_error = cybertron::FDT_ERROR_NETWORK;
                                task->_stage = cybertron::FDT_STAGE_WAITING_FOR_RETRY;
                                break;
                        }
                        task->callback_status(true);
                        curl_multi_remove_handle(curl_multi, msg->easy_handle);
                        curl_easy_cleanup(msg->easy_handle);
                        task->_curl = nullptr;


                        if (task->_last_error == cybertron::FDT_ERROR_OK) {
                            if (task->_stage == cybertron::FDT_STAGE_COMPLETED) {
                                _downloading_tasks.erase(task);
                            } else if (task->_stage == cybertron::FDT_STAGE_WAITING_FOR_UNZIP) {
                                Task *t = new Task(task->_url);
                                t->_dst_dir = task->_dst_dir;
                                t->_dst_path = task->_dst_path;
                                _wait_for_unzip_tasks_mutex.lock();
                                _wait_for_unzip_tasks.insert(std::make_pair(time(NULL), t));
                                _wait_for_unzip_tasks_mutex.unlock();
                                _downloading_tasks.erase(task);
                            } else {
                                run_task(task, curl_multi);
                            }
                        } else { //fail
                            _downloading_tasks.erase(task);
                            _wait_for_retry_tasks.insert(std::make_pair(time(NULL) + 10, task));
                        }
                    }
                } while (msgs_left);
            }
        }

        static Manager *_instance;

        int _max_concurrent_task_num;
        cybertron::FDT_STATUS_CB _cb;
        bool _quit;

        std::unordered_map<std::string, Task *> _tasks;  //tasks = running + queuing + wait_for_retry + completed
        std::unordered_set<Task *> _downloading_tasks;
        std::multimap<time_t, Task *> _queuing_tasks;
        std::multimap<time_t, Task *> _wait_for_retry_tasks;  //the key is the time to retry

        std::thread _download_thread;
        std::thread _unzip_thread;

        std::unordered_map<std::string, cybertron::FDT_STATUS> _task_status;
        std::mutex _task_status_mutex;

        std::queue<Cmd> _cmd_queue;
        std::mutex _cmd_queue_mutex;

        std::multimap<time_t, Task *> _wait_for_unzip_tasks;
        std::mutex _wait_for_unzip_tasks_mutex;
    };

    Manager *Manager::_instance = new Manager();
}

CYBERTRON_BEGIN

    int FileDownloaderTaskManager::start(int max_concurrent_task_num, FDT_STATUS_CB callback) {
        return FDTM_IMPL::Manager::instance()->start(max_concurrent_task_num, callback);
    }

    int FileDownloaderTaskManager::stop() {
        return FDTM_IMPL::Manager::instance()->stop();
    }

    int FileDownloaderTaskManager::add_task(std::string url, std::string md5sum, std::string dst_dir, bool decompress,
                                            FDT_STATUS_CB callback) {
        return FDTM_IMPL::Manager::instance()->add_task(url, md5sum, dst_dir, decompress, callback);
    }

    int FileDownloaderTaskManager::delete_task(std::string url) {
        return FDTM_IMPL::Manager::instance()->delete_task(url);
    }

    int FileDownloaderTaskManager::task_status(std::string url, FDT_STATUS &status) {
        return FDTM_IMPL::Manager::instance()->task_status(url, status);
    }

    const char *FileDownloaderTaskManager::stage_string(FDT_STAGE stage) {
        switch (stage) {
            case cybertron::FDT_STAGE_QUEUING:
                return "stage_queuing";
            case cybertron::FDT_STAGE_DONWLOADING:
                return "stage_downloading";
            case cybertron::FDT_STAGE_WAITING_FOR_RETRY:
                return "stage_waiting_for_retry";
            case cybertron::FDT_STAGE_WAITING_FOR_UNZIP:
                return "stage_waiting_for_unzip";
            case cybertron::FDT_STAGE_UNZIPING:
                return "stage_unziping";
            case cybertron::FDT_STAGE_FAILED:
                return "stage_failed";
            case cybertron::FDT_STAGE_DELETED:
                return "stage_deleted";
            case cybertron::FDT_STAGE_COMPLETED:
                return "stage_completed";
            default:
                return "invalid_stage";
        }
    }

    const char *FileDownloaderTaskManager::error_string(FDT_ERROR err) {
        switch (err) {
            case cybertron::FDT_ERROR_OK:
                return "error_ok";
            case cybertron::FDT_ERROR_NETWORK:
                return "error_network";
            case cybertron::FDT_ERROR_DUPLICATE:
                return "error_duplicate";
            case cybertron::FDT_ERROR_FILE:
                return "error_file";
            case cybertron::FDT_ERROR_SERVER:
                return "error_server";
            case cybertron::FDT_ERROR_UNZIP:
                return "error_unzip";
            case cybertron::FDT_ERROR_UNKNOWN:
                return "error_unknown";
            default:
                return "invalid_error";
        }

    }


CYBERTRON_END
